package com.ck.core.lock;

import io.lettuce.core.SetArgs;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;

/**
 * 
 * <p>Title: RedisDistributeLock</p>  
 * <p>Description: DistributeLock implements by redis</p>  
 * @author huangf 
 * @date 2018年7月4日
 */
@Component
public class RedisDistributeLock {
    private Logger logger = LoggerFactory.getLogger(getClass().getSimpleName());

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    public RedisLock getLock(final String key, long expireSeconds) {
        return getLock(key, expireSeconds, 0, 0);
    }

    /**
     * 获取redis的分布式锁，内部实现使用了redis的setnx。如果锁定失败返回null，如果锁定成功则返回RedisLock对象，调用方需要调用RedisLock.unlock()方法来释放锁
     * <br/>
     * <span style="color:red;">此方法在获取失败时会自动重试指定的次数,由于多次等待会阻塞当前线程，请尽量避免使用此方法</span>
     *
     * @param key                     要锁定的key
     * @param expireSeconds           key的失效时间(秒):ex单位秒，px单位毫秒
     * @param maxRetryTimes           最大重试次数,如果获取锁失败，会自动尝试重新获取锁；
     * @param retryIntervalTimeMillis 每次重试之前sleep等待的毫秒数
     * @return 获得的锁对象（如果为null表示获取锁失败），后续可以调用该对象的unlock方法来释放锁.
     */
    public RedisLock getLock(final String key, final long expireSeconds, int maxRetryTimes, long retryIntervalTimeMillis) {
        final String value = key.hashCode() + "";

        int maxTimes = maxRetryTimes + 1;
        for (int i = 0; i < maxTimes; i++) {
            String status = stringRedisTemplate.execute(new RedisCallback<String>() {
                @Override
                public String doInRedis(RedisConnection connection) throws DataAccessException {
                    Object nativeConnection = connection.getNativeConnection();
                    String redisResult = "";
                    byte[] keyByte = rawKey(key);
                    byte[] valueByte = rawValue(value);
                    if (nativeConnection instanceof RedisAsyncCommands) {
                        RedisAsyncCommands commands = (RedisAsyncCommands) nativeConnection;
                        //同步方法执行、setnx禁止异步
                        redisResult = commands
                                .getStatefulConnection()
                                .sync()
                                .set(keyByte, valueByte, SetArgs.Builder.nx().ex(expireSeconds));
                    } else if (nativeConnection instanceof RedisAdvancedClusterAsyncCommands) {
                        RedisAdvancedClusterAsyncCommands clusterAsyncCommands = (RedisAdvancedClusterAsyncCommands) nativeConnection;
                        redisResult = clusterAsyncCommands
                                .getStatefulConnection()
                                .sync()
                                .set(keyByte, keyByte, SetArgs.Builder.nx().ex(expireSeconds));
                    }
                    return redisResult;
                }
            });
            //抢到锁
            if ("OK".equalsIgnoreCase(status)) {
                return new RedisLockInner(stringRedisTemplate, key, value);
            }

            if (retryIntervalTimeMillis > 0) {
                try {
                    Thread.sleep(retryIntervalTimeMillis);
                } catch (InterruptedException e) {
                    logger.error(e.getMessage(),e);
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            if (Thread.currentThread().isInterrupted()) {
                break;
            }
        }

        return null;
    }

    /**
     * 序列化key为byet[]
     *
     * @param key
     * @return
     */
    private byte[] rawKey(Object key) {
        if (keySerializer() == null && key instanceof byte[]) {
            return (byte[]) key;
        }

        return keySerializer().serialize(key);
    }

    /**
     * 获取key序列化器
     *
     * @return
     */
    private RedisSerializer keySerializer() {
        return stringRedisTemplate.getKeySerializer();
    }


    /**
     * 序列化value
     *
     * @param value
     * @return
     */
    private byte[] rawValue(Object value) {
        if (keySerializer() == null && value instanceof byte[]) {
            return (byte[]) value;
        }
        return valueSerializer().serialize(value);
    }

    /**
     * 获取value序列化器
     *
     * @return
     */
    private RedisSerializer valueSerializer() {
        return stringRedisTemplate.getValueSerializer();
    }

    private class RedisLockInner implements RedisLock {
        private StringRedisTemplate stringRedisTemplate;
        private String key;
        private String expectedValue;

        protected RedisLockInner(StringRedisTemplate stringRedisTemplate, String key, String expectedValue) {
            this.stringRedisTemplate = stringRedisTemplate;
            this.key = key;
            this.expectedValue = expectedValue;
        }

        /**
         * 释放redis分布式锁
         */
        @Override
        public void unlock() {
            List<String> keys = Collections.singletonList(key);
            stringRedisTemplate.delete(keys);
        }

        @Override
        public void close() throws Exception {
            this.unlock();
        }
    }

}
